package com.example.flow

import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.Test


/**
 * 测试Flow
 * 异步方式创建多个值的集合
 * Flow默认是在挂起协程中处理
 * Flow流特性：
 * Flow是冷流。Flow构建器中的代码直到流被收集的时候才运行。只有消费的时候才会创建Flow
 *
 * 流使用emit发射元素
 * 流使用collect收集元素
 *
 * @author zhouronghua
 * @time 2022/1/5 12:42 下午
 */
class FlowCreateTest {

    /**
     * 创建List
     * List是同步定长型集合
     *
     * @author zhouronghua
     * @time 2022/1/5 12:52 下午
     */
    suspend fun sampleList(): List<Int> {
        // 每隔一秒发射一个元素
        delay(1000)
        return listOf(1, 2, 3)
    }


    /**
     * 创建Flow流
     *
     * @author zhouronghua
     * @time 2022/1/5 12:52 下午
     */
    fun sampleFlow(): Flow<Int> = flow<Int> {
        for (i in 1..3) {
            // 每隔一秒发射一个元素
            delay(1000)
            emit(i)
        }
    }

    fun simple(): Flow<Int> = flow { // 流构建器
        for (i in 1..3) {
            delay(100) // 假装我们在这里做了一些有用的事情
            emit(i) // 发送下一个值
            println("发射 $i ${Thread.currentThread().name}")
        }
    }

    @InternalCoroutinesApi
    @Test
    fun test_list_multi_value() = runBlocking<Unit> {
        sampleList().forEach { it ->
            println(it)
        }
    }

    /**
     * Flow流收集处理是挂起处理不阻塞主线程
     * 说明：当没有切换上下文的时候，发射流和收集流是在同一个协程。
     *
     * 执行结果：
     * 发射 2 Test worker @coroutine#1
     * I'm not blocked 3
     * receive 3 Test worker @coroutine#1
     * @author zhouronghua
     * @time 2022/1/5 8:15 下午
     */
    @Test
    fun test_flow_multi_value() = runBlocking<Unit> {
        // 启动并发的协程以验证主线程并未阻塞
        launch {
            for (k in 1..3) {
                println("I'm not blocked $k")
                delay(100)
            }
        }
        // 收集这个流
        simple().collect { value ->
            println("receive $value ${Thread.currentThread().name}")
        }
    }

    /**
     * Flow是冷流
     * 收集的时候执行流发射。
     * 因此第二轮收集的时候，流发射再次会执行。
     *
     * 冷启动：现学现卖
     * 热流：创业，之前有积累。不管你是否接收，发射端主动发送，存入缓存区
     *
     * @author zhouronghua
     * @time 2022/1/5 8:22 下午
     */
    @Test
    fun test_flow_is_cold() = runBlocking {
        println("第一轮收集")
        // 收集这个流
        simple().collect { value ->
            println("receive $value ${Thread.currentThread().name}")
        }
        println("第二轮收集")
        simple().collect { value ->
            println("第二轮 receive $value ${Thread.currentThread().name}")
        }
    }

    /**
     * 流的连续性
     * 说明：收集时元素是按照发射端的顺序收集
     * 流的每次收集都是按照顺序执行的，除非使用特殊操作符
     * 从上游到下游每个过度操作符都会处理每个发射出的元素，然后再交给末端操作符
     *
     * @author zhouronghua
     * @time 2022/1/5 8:25 下午
     */
    @Test
    fun test_flow_is_continuation() = runBlocking {
        (1..5).asFlow()
            .filter {
                // 上游1..5共5个元素
                println("filter $it")
                // 过滤偶数值元素
                it % 2 == 0
            }.map {
                // 上游的偶数进行map
                println("map $it")
                "to string $it"
            }.collect {
                println("collect $it")
            }
    }

    /**
     * 测试流的构建器
     * 1）flowOf可以构建一个发射固定值的流
     * 2）asFlow可以将序列和各种集合转换为流
     *
     * @author zhouronghua
     * @time 2022/1/5 10:07 下午
     */
    @Test
    fun test_flow_creator() = runBlocking {
        // 发射5个元素的流
        flowOf(1, 2, 3, 4, 5).collect {
            println("Collect $it")
        }
        // 发射一个列表
        println("发射一个列表")
        flowOf(listOf(1, 2, 3, 4, 5)).collect {
            println("Collect $it")
        }
        println("将List作为流发射")
        listOf(1, 2, 3, 4, 5).asFlow().collect {
            println("Collect $it")
        }
    }
}

